/*
 * Scheduler.c
 *
 *  Created on: 2011-8-14
 *      Author: ryan
 */
#include "Scheduler.h"

int SIZE, RANK;
int pre, next;

void sendTask(KEY_VALUE *task, int dst);
void recvTask(KEY_VALUE *task);

void SEND(void* buf, int size, TYPE t, int dest){
	if(dest == PRE){
		MPI_CHECK(MPI_Isend(buf, size, t, pre, COMTAG, MPI_COMM_WORLD, &request));
	}else if(dest == NEXT){
		MPI_CHECK(MPI_Isend(buf, size, t, next, COMTAG, MPI_COMM_WORLD, &request));
	}else if(dest == 0){
		MPI_CHECK(MPI_Isend(buf, size, t, RANK, COMTAG, MPI_COMM_WORLD, &request));
	}
}

void RECV(void* buf, int size, TYPE t, int src){
	if(src == PRE){
		MPI_CHECK(MPI_Recv(buf, size, t, pre, COMTAG, MPI_COMM_WORLD, &status));
	}else if(src == NEXT){
		MPI_CHECK(MPI_Recv(buf, size, t, next, COMTAG, MPI_COMM_WORLD, &status));
	}else if(src == 0){
		MPI_CHECK(MPI_Recv(buf, size, t, RANK, COMTAG, MPI_COMM_WORLD, &status));
	}
}

void myAbort(){
	MPI_Abort(MPI_COMM_WORLD, -1);
}

void submit(int argc, char **argv) {
	int i;
	MPI_CHECK(MPI_Init(&argc, &argv));
	MPI_CHECK(MPI_Comm_size(MPI_COMM_WORLD, &SIZE));
	MPI_CHECK(MPI_Comm_rank(MPI_COMM_WORLD, &RANK));
	KEY_VALUE *task;
	if (RANK == 0) {		//master node.
		TASK tasks;
		splitTask(argc, argv, &tasks);
		task = &tasks.subtasks[0];
		for(i = 1; i < tasks.num; ++i){
			pre = i - 1;
			next = i + 1;
			MPI_CHECK(MPI_Isend(&pre, 1, MPI_INT, i, COMTAG, MPI_COMM_WORLD, &request));
			MPI_CHECK(MPI_Isend(&next, 1, MPI_INT, i, COMTAG, MPI_COMM_WORLD, &request));
			sendTask(&tasks.subtasks[i], i);
		}
		pre = -1;
		next = 1;
	} else {
		MPI_CHECK(MPI_Recv(&pre, 1, MPI_INT, 0, COMTAG, MPI_COMM_WORLD, &status));
		MPI_CHECK(MPI_Recv(&next, 1, MPI_INT, 0, COMTAG, MPI_COMM_WORLD, &status));
		task = malloc(sizeof(KEY_VALUE));
		recvTask(task);
	}
	map(task, NULL);
	stop();
}

void stop(){
	MPI_CHECK(MPI_Finalize());
}

void sendTask(KEY_VALUE *task, int dst){
	MPI_CHECK(MPI_Isend(&task->keySize, 1, MPI_INT, dst, COMTAG, MPI_COMM_WORLD, &request));
	MPI_CHECK(MPI_Isend(task->key, task->keySize, MPI_BYTE, dst, COMTAG, MPI_COMM_WORLD, &request));

	MPI_CHECK(MPI_Isend(&task->valueSize, 1, MPI_INT, dst, COMTAG, MPI_COMM_WORLD, &request));
	MPI_CHECK(MPI_Isend(task->value, task->valueSize, MPI_BYTE, dst, COMTAG, MPI_COMM_WORLD, &request));
}
void recvTask(KEY_VALUE *task){
	MPI_CHECK(MPI_Recv(&task->keySize, 1, MPI_INT, 0, COMTAG, MPI_COMM_WORLD, &status));
	task->key = malloc(task->keySize);
	MPI_CHECK(MPI_Recv(task->key, task->keySize, MPI_BYTE, 0, COMTAG, MPI_COMM_WORLD, &status));

	MPI_CHECK(MPI_Recv(&task->valueSize, 1, MPI_INT, 0, COMTAG, MPI_COMM_WORLD, &status));
	task->value = malloc(task->valueSize);
	MPI_CHECK(MPI_Recv(task->value, task->valueSize, MPI_BYTE, 0, COMTAG, MPI_COMM_WORLD, &status));
}
